package com.dmp

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, SaveMode}

/**
  * ClassName:AreaAnalyseRpt
  * Package:com.dmp
  * Desciption:
  *
  * @date:2019 /8/23 14:29
  * @author:17611219021 @sina.cn
  */
object AreaAnalyseRpt {
  def main(args: Array[String]): Unit = {
    if(args.length !=2){
      println(
        """
          |com.dmp.fileToParquet
          |参数：
          | inputpath
          | outputPath
        """.stripMargin)
      sys.exit()
    }
    val conf: SparkConf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    //指定keyo序列化器与snappy格式，默认是gzip格式
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.parquet.compression.codec","snappy")
    val sc: SparkContext = new SparkContext(conf)
    val Array(inputpath,outputPath)= args
    val sqlContext: SQLContext = new SQLContext(sc)

    val df1 = sqlContext.read.parquet(inputpath)
    df1.registerTempTable("tb_areaanalyse")
    val df2 = sqlContext.sql(
      """
        select province,area,
        sum( case when requestmode = 1 and processnode >= 1 then 1 else 0 end) 原始请求,
        sum( case when requestmode = 1 and processnode >= 2 then 1 else 0 end ) 有效请求,
        sum( case when requestmode = 1 and processnode = 3 then 1 else 0 end ) 广告请求,
        sum( case when iseffective = 1 and isbiding = 1 and isbid = 1 and adorderid !=0 then 1 else 0 end ) 参与竞价数,
        sum( case when iseffective = 1 and isbiding = 1 and iswin= 1 then 1 else 0 end ) 竞价成功数,
        sum( case when iseffective = 1 and isbiding = 1 and iswin= 1 then winprice else 0 end ) 广告消费,
        sum( case when iseffective = 1 and isbiding = 1 and iswin= 1 then adpayment else 0 end ) 广告成本,
        sum( case when requestmode = 2 and iseffective = 1 then 1 else 0 end ) 展示数,
        sum( case when requestmode = 3 and iseffective = 1 then 1 else 0 end ) 点击数
        from tb_areaanalyse group by province,area
      """.stripMargin)
    df2.show()
    //注意，字段是中文的，写mysql会报错，请将字段名字改为英文
    val load = ConfigFactory.load()
    val properties = new Properties()
    properties.setProperty("user",load.getString("jdbc.user"))
    properties.setProperty("password",load.getString("jdbc.password"))
    //在表尾部追加数据
    df2.write.mode(SaveMode.Append).jdbc(load.getString("jdbc.url"),load.getString("jdbc.tabname2"),properties)
    sc.stop()

  }

}
